
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpRequest.BodyPublishers;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodyHandlers;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.function.Consumer;

import org.apache.mesos.v1.master.Protos;
import org.apache.mesos.v1.master.Protos.Event;
import org.apache.mesos.v1.master.Protos.Response.GetState;

/**
 * MESOS Event Demo With PROTOBUF, New HttpClient API In Java11
 *
 * @author xiaowei
 *
 */
public class EventDemo {
	private String mesosUrl;

	public EventDemo(String mesosUrl) {
		this.mesosUrl = mesosUrl;
	}

	public void callMesos() {

		Protos.Call subscribedCall = Protos.Call.newBuilder()
				.setType(Protos.Call.Type.SUBSCRIBE).build();
		byte[] payload = subscribedCall.toByteArray();

		HttpClient client = HttpClient.newHttpClient();

		HttpRequest request = HttpRequest.newBuilder().uri(URI.create(mesosUrl))
				.timeout(Duration.ofMinutes(2))
				.header("Content-Type", "application/x-protobuf")
				.header("Accept", "application/x-protobuf")
				.header("Message-Accept", "application/recordio")
				.POST(BodyPublishers.ofByteArray(payload)).build();

		final EventDemo ed = this;

		client.sendAsync(request, BodyHandlers.ofInputStream())
				.thenApply(HttpResponse::body)
				.thenAccept(new Consumer<InputStream>() {

					@Override
					public void accept(InputStream in) {

						ByteBuffer buffer = ByteBuffer.allocate(104857600); // 100MiB Just a example

						try {

							for (int i = 0; i < 10; i++) {
								buffer.clear();
								ed.readRecord(in, buffer);
								System.out.printf(
										"\n%s we get the %d record, length of: %s\n",
										LocalDateTime.now(), i, buffer.limit());
								Event ev = ed.parseRecord(buffer);
								if (ev != null) {
									ed.handleEvent(ev);
								}

								try {
									Thread.sleep(100);
								} catch (InterruptedException e) {
								}

							}

						} catch (IOException e) {
							e.printStackTrace();
						}

					}

				}).join();

	}

	protected void handleEvent(Event e) {
		showEvent(e);
	}

	public void showEvent(Event e) {
		System.out.println("This is a " + e.getType() + " Event");
		switch (e.getType().getNumber()) {
		case Event.Type.SUBSCRIBED_VALUE:
			showSubscribedEvent(e);
			break;
		case Event.Type.TASK_ADDED_VALUE:
			showTaskAddedEvent(e);
			break;
		case Event.Type.TASK_UPDATED_VALUE:
			showTaskUpdatedEvent(e);
			break;
		}

	}

	public void showSubscribedEvent(Event e) {
		String showtpl = "The cluster has %s agents, %d frameworks, %s tasks\n";
		GetState state = e.getSubscribed().getGetState();
		System.out.printf(showtpl, state.getGetAgents().getAgentsCount(),
				state.getGetFrameworks().getFrameworksCount(),
				state.getGetTasks().getTasksCount());

	}

	public void showTaskAddedEvent(Event e) {
		System.out.printf("A new task %s was %s\n",
				e.getTaskAdded().getTask().getTaskId().getValue(),
				e.getTaskAdded().getTask().getState());
	}

	public void showTaskUpdatedEvent(Event e) {
		System.out.printf("The task %s was %s\n",
				e.getTaskUpdated().getStatus().getTaskId().getValue(),
				e.getTaskUpdated().getState());
	}

	protected Event parseRecord(ByteBuffer buffer) throws IOException {
		return Event.parseFrom(buffer);
	}

	protected ByteBuffer readRecord(InputStream in, ByteBuffer buffer)
			throws IOException {

		// first read head of the record
		for (int i = 0; i < 20; i++) {
			int b = in.read();
			if (b != '\n') {
				buffer.put((byte) b);
			} else {
				break;
			}
		}

		if (buffer.position() == 20) {
			throw new IOException(
					"can not find record second part, the `LF` Character: "
							+ new String(buffer.array(), 0, buffer.position()));
		}

		// then read the third of the record
		int recordBodySize = Integer
				.parseInt(new String(buffer.array(), 0, buffer.position()));

		if (recordBodySize <= 0 || recordBodySize > buffer.capacity()) {
			throw new IOException(
					"record data, length error: " + recordBodySize);
		}
		buffer.clear();
		int read = in.readNBytes(buffer.array(), 0, recordBodySize);
		if (read != recordBodySize) {
			throw new IOException("record data not enough, need "
					+ recordBodySize + ", only have: " + read);
		}
		buffer.position(0);
		buffer.limit(recordBodySize);
		return buffer;
	}

	public static void main(String[] args) {

		String url = "http://10.0.209.3:5050/api/v1";
		if (args.length > 1) {
			url = args[1];
		}

		System.out.printf("%s we subscribe the mesos: %s\n", LocalDateTime.now(), url);

		EventDemo me = new EventDemo(url);

		me.callMesos();

	}

}